package com.timing.finder.framework;

import com.example.autojob.logging.model.producer.AutoJobLogHelper;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * 时序数据处理器
 *
 * @author JingGe(* ^ ▽ ^ *)
 * @date 2023-03-03 13:14
 * @email 1158055613@qq.com
 */
@Slf4j
public abstract class TimeSeriesDataHandler<T extends TimeSeriesData> {
    /**
     * 返回该组数据的key，一组数据的key应该唯一
     *
     * @param data 时序数据
     * @return java.lang.String
     * @author JingGe(* ^ ▽ ^ *)
     * @date 2023/3/3 14:04
     */
    public abstract String getKey(List<T> data);

    public abstract long startOffset();

    public abstract long unitOffset();

    public abstract long offsetIndex(T data);

    public abstract long initialSize();

    public abstract CacheRepository cacheRepository();

    public abstract long expiringTime();

    int storeInCache(List<T> seriesData) {
        if (seriesData == null || seriesData.size() == 0) {
            return 0;
        }
        String key = cacheRepository().formatKey(getKey(seriesData));
        if (cacheRepository().createKeyListIfAbsent(key, startOffset(), initialSize(), expiringTime(), TimeUnit.SECONDS)) {
            Map<Long, T> valuesMap = new HashMap<>();
            seriesData.forEach(item -> valuesMap.put(offsetIndex(item), item));
            return cacheRepository().indexSets(key, valuesMap);
        }
        return 0;
    }

    public long clear(String pattern) {
        return cacheRepository().deletePattern(pattern);
    }

    int storeInCache(Map<String, List<T>> dataMap) {
        AutoJobLogHelper logHelper = AutoJobLogHelper.getInstance();
        logHelper.setSlf4jProxy(log);
        Map<String, Long> keyMap = new HashMap<>();
        List<ValueEntry<T>> entries = new ArrayList<>();
        for (Map.Entry<String, List<T>> entry : dataMap.entrySet()) {
            keyMap.put(cacheRepository().formatKey(entry.getKey()), initialSize());
            entry
                    .getValue()
                    .forEach(item -> entries.add(new ValueEntry<>(cacheRepository().formatKey(entry.getKey()), offsetIndex(item), item)));

        }
        int count = 0;
        try {
            count = cacheRepository().loadBatch(keyMap, entries, startOffset(), expiringTime(), TimeUnit.SECONDS);
        } catch (DataSaveException e) {
            logHelper.warn("{}", e.getMessage());
            logHelper.info("开始扫描{}个桶的存在性", keyMap.size());
            long start = System.currentTimeMillis();
            Map<String, Boolean> exist = null;
            try {
                exist = cacheRepository().scanKeysExist(new ArrayList<>(keyMap.keySet()));
            } catch (ScanBucketException ex) {
                logHelper.error("扫描桶时发生异常：{}", ex.getMessage());
                throw new BucketCreateException("桶创建失败");
            }
            logHelper.info("扫描存在性完成，用时{}ms", System.currentTimeMillis() - start);
            Map<String, Long> unExistBuckets = new HashMap<>();
            for (Map.Entry<String, Boolean> entry : exist.entrySet()) {
                if (!entry.getValue()) {
                    unExistBuckets.put(entry.getKey(), initialSize());
                }
            }
            logHelper.warn("存在{}个桶没被创建", unExistBuckets.size());
            if (cacheRepository().createKeyListsIfAbsent(unExistBuckets, startOffset(), expiringTime(), TimeUnit.SECONDS)) {
                logHelper.info("成功创建不存在的桶");
            } else {
                throw new BucketCreateException("桶创建失败");
            }
            throw e;
        }
        return count;
    }


}
